

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Vector;


/**
 * 
 * @author Mikkin Patel
 *
 */
public class FileSplitService extends Thread {
	private String fileName;
	private BufferedReader readFile;

	public FileSplitService(String fileName){
		this.fileName = fileName;
	}
	
	public void distributeChunksToDataServers() throws IOException{
		ObjectInputStream in;
		ObjectOutputStream out;
		SocketChannel dataServerConnectionSocket;
		String dataServerName;
		DataServerInfo dsInfo;
		Vector<String> chunkList = ControlServer.fileToChunkMap.get(fileName);

		int numLoops = (chunkList.size() / ControlServer.dataServerList.size()) + 1;
		
		int  chunkListCounter = 0;
		for( int i = 0; i < numLoops; i++){
			Iterator<String> dsIterator = ControlServer.dataServerList.keySet().iterator();
			while( dsIterator.hasNext() && chunkListCounter < chunkList.size()){
				dsInfo = ControlServer.dataServerList.get(dsIterator.next()).get(0);
				dataServerConnectionSocket = dsInfo.getDataSocket();
				String chunkName = chunkList.get(chunkListCounter);
				dsInfo.getObjOutputStream().writeObject(chunkName);
				FileInputStream chunkOut = new FileInputStream(chunkName);
				FileChannel chunkChannel = chunkOut.getChannel();
				dsInfo.getObjOutputStream().writeObject(chunkChannel.size());
				chunkChannel.transferTo(0, chunkChannel.size(), dataServerConnectionSocket);
				
				chunkChannel.close();
				chunkOut.close();
				
				File f = new File(chunkName);
				if(f.exists()){
					System.out.println(chunkName + " deleted from control server");
					f.delete();
				}
				
				//dsInfo.getObjOutputStream().writeObject(new File(chunkList.get(chunkListCounter)));
				chunkListCounter++;
			}
		}
	}

	public void run(){
		try {
			readFile = new BufferedReader(new FileReader(fileName));
			String line = readFile.readLine();
			System.out.println(line);
			line = readFile.readLine();
			String timeLine=null;
			line = readFile.readLine();
			//System.out.println(line);
			String[] lineArray = line.split(" ");
			Vector<String> chunkNames = new Vector<String>();

			while((line) != null){
				String fName = lineArray[1]+"_"+fileName;
				
				if(ControlServer.fileToChunkMap.get(fileName)== null){
					chunkNames.addElement(fName);
					ControlServer.fileToChunkMap.put(fileName,chunkNames);
				}else{
					chunkNames = ControlServer.fileToChunkMap.get(fileName);
					chunkNames.addElement(fName);
					ControlServer.fileToChunkMap.put(fileName,chunkNames);
				}
				
				File chunkFile = new File(fName);
				timeLine=lineArray[1];
				try {
					BufferedWriter buff = new BufferedWriter(new FileWriter(chunkFile));
					while( line != null && timeLine.equals(lineArray[1])){
						buff.write(line);
						line = readFile.readLine();
						if(line == null){
							break;
						}else{
							lineArray = line.split(" ");
						}
					}
					buff.close();
					
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			//System.out.println(ControlServer.fileToChunkMap);
			distributeChunksToDataServers();
			
		} catch (Exception e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
	}
}
